package com.study.datasource.crawler;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.study.datasource.bean.WaterQuality;
import com.study.datasource.util.TimeUtils;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @Author xc
 * @Date 2023/4/8 23:01
 * @Version 1.0
 */

/*
 * 暂时没用上。。。。。。。。
 */
@Component
public class WaterQualityData {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    // 后续需要将该方法改为定时任务，如每天8点定时爬取疫情数据
    //springboot的定时任务工具
    //@Scheduled(initialDelay = 1000,fixedDelay = 1000)
    //@Scheduled(cron = "0/1 * * * * ?")//每隔一秒执行一次 0 0/1 * * * ?（一分钟）
    //@Scheduled(cron = "0 0 8 * * ?")//每天8点定时执行
    @Scheduled(initialDelay = 1000, fixedDelay = 1000 * 60 * 60 * 24)//首次任务执行的延迟时间和任务之间的间隔时间
//    @Scheduled(cron = "0/5 * * * * ?")//每隔5秒
    public void testCrawling() throws Exception {
        //System.out.println("每个五秒执行一次");
        String datetime = TimeUtils.format(System.currentTimeMillis(), "yyyy-MM-dd");

        final String datatime = TimeUtils.format(System.currentTimeMillis(), "yyyy-MM-dd");

        /*
        数据处理
         */
        // 1、加载HTML文件
        Document doc = Jsoup.connect("https://moonapi.com/WaterQuality/RealTimeWQ/index.html").get();
        // 2、数据处理

        // 获取表格元素
        Element table = doc.select("#wq-table").get(0);
        // 获取表头
        Elements rows = table.select("tr");
        Element rowTop = rows.get(0);
        Elements ths = rowTop.select("td");

        List<String> headers = new ArrayList<>();
        for (Element th : ths) {
            headers.add(th.text());
        }

        // 获取表格数据
        Elements trs = table.select("tbody").select("tr");
        List<Map<String, String>> data = new ArrayList<>();
        for (Element tr : trs) {
            Map<String, String> row = new HashMap<>();
            Elements tds = tr.select("td");
            int i = 0;
            for (Element td : tds) {
                String text = td.text();
                if (text.equals("*")) {
                    text = "0";
                }
                if (i < headers.size()) {
                    row.put(headers.get(i), text);
                    i++;
                }
            }
            data.add(row);
        }

        // 3、将数据转换为JSON格式，导入gson
        String json = new Gson().toJson(data);
        System.out.println(json);

        // 4、进一步解析JSON
        List<WaterQuality> waterQualities = new ArrayList<>();
        JSONArray jsonArray = JSON.parseArray(json);
        for (int i = 0; i < jsonArray.size(); i++) {
            JSONObject jsonObject = jsonArray.getJSONObject(i);
            WaterQuality waterQuality = new WaterQuality();
            waterQuality.setMid(i + 1);
            waterQuality.setWaterTemperature(Float.parseFloat(jsonObject.getString("水温(℃)")));
            waterQuality.setTurbidity(Float.parseFloat(jsonObject.getString("浊度(NTU)")));
            waterQuality.setChlorophyllAlpha(Float.parseFloat(jsonObject.getString("叶绿素α(mg/L)")));

            waterQuality.setWatershedRiver(jsonObject.getString("流域河流"));
            waterQuality.setProvinceCity(jsonObject.getString("省市"));

            waterQuality.setDissolvedOxygen(Float.parseFloat(jsonObject.getString("溶解氧(mg/L)")));

            waterQuality.setMonitoringTime(jsonObject.getString("监测时间"));
            // waterQuality.setDissolvedOxygen(Date.parse(jsonObject.getString("监测时间")));

            waterQuality.setTotalPhosphorus(Float.parseFloat(jsonObject.getString("总磷(mg/L)")));
            waterQuality.setAlgalDensity(Float.parseFloat(jsonObject.getString("藻密度(cells/L)")));
            waterQuality.setConductivity(Float.parseFloat(jsonObject.getString("电导率(μS/cm)")));
            waterQuality.setTotalNitrogen(Float.parseFloat(jsonObject.getString("总氮(mg/L)")));
            waterQuality.setPH(Float.parseFloat(jsonObject.getString("pH(无量纲)")));

            waterQuality.setSectionPoint(jsonObject.getString("断面位点"));

            waterQuality.setAmmoniaNitrogen(Float.parseFloat(jsonObject.getString("氨氮(mg/L)")));

            waterQuality.setWaterQualityType(jsonObject.getString("水质 类别"));

            waterQuality.setPermanganateIndex(Float.parseFloat(jsonObject.getString("高锰酸盐指数(mg/L)")));


            waterQuality.setSiteCondition(jsonObject.getString("站点 情况"));


            waterQualities.add(waterQuality);
        }

        //打印对应的实体类数据
        for (WaterQuality waterQuality : waterQualities) {
            System.out.println(waterQuality);
            String jsonString = JSON.toJSONString(waterQuality);
            // System.out.println(jsonString);
            kafkaTemplate.send("sewageDischarge", waterQuality.getMid(), jsonString);
        }
        System.out.println("已经完成对应实体的发送");
        Thread.sleep(1000000);//线程阻塞


    }
}
